package kafka

import (
	"github.com/Shopify/sarama"
	log "github.com/sirupsen/logrus"
	"sync"
)

type singleton struct {
	enable   bool
	producer sarama.SyncProducer
}

var ins *singleton
var once sync.Once
var mu sync.Mutex

func GetInstance() *singleton {
	once.Do(func() {
		ins = &singleton{}
	})
	return ins
}

func (c *singleton) Enabled() bool {
	return c.enable
}

func (c *singleton) ConnectKafka(brokers []string) {
	mu.Lock()
	defer mu.Unlock()

	config := sarama.NewConfig()
	config.Metadata.Retry.Max = 3
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Return.Successes = true
	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.Initial = sarama.OffsetOldest

	producer, err := sarama.NewSyncProducer(brokers, config)
	if err != nil {
		log.Errorf("new sync producer err:%+v", err)
	}

	c.enable = true
	c.producer = producer
}

func (c *singleton) DisConnectKafka() {
	mu.Lock()
	defer mu.Unlock()
	c.enable = false
}

func (c *singleton) PublishMessage(data string) error {
	if !c.enable {
		return nil
	}
	return nil
}
